/*
* Copyright (C) 2020 ~ 2024 chaigec All Rights Reserved.
*
* Author:     chaigec <chaigec@163.com>
* Maintainer: chaigec <chaigec@163.com>

 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
*/

#include "MkGbMediaEndUser.h"
#include "MkEvent/MkEventPool.h"
#include "MkRtp/MkRtpUdpServer.h"
#include "MkRtp/MkRtpTcpClient.h"
#include "MkGbConsumer.h"
#include "MkEvent/MkEvent.h"
#include "MkUtil/MkLog.h"
#include "MkUtil/MkUtil.h"

CMkGbMediaEndUser::CMkGbMediaEndUser(CMkEventPool* pEventPool, CMkTimer* pTimer)
    : CMkGbEndUser(pEventPool, pTimer)
{

}

Uint32 CMkGbMediaEndUser::Start(const MkGbStartParam& Param, Uint16 RtpBeginPort)
{
    Uint32 ErrorCode = NoneError;
    m_RtpBeginPort = RtpBeginPort;
    MkVector<CMkEvent*>::iterator it = GetEventPool()->GetVecEvent().begin();
    Uint16 OffsetPort = RtpBeginPort;
    for (; it != GetEventPool()->GetVecEvent().end(); it++) {
        do {
            CMkRtpUdpServer *pUdpServer = new CMkRtpUdpServer(*it, GetTimer());
            if (pUdpServer && NoneError == pUdpServer->Start(OffsetPort)) {
                m_VecRtpUdpServer.emplace_back(pUdpServer);
                break;
            }
            OffsetPort += 2;
        } while (1);
    }
    MkInfoLog("start rtp pool begin port:%d  success\n", RtpBeginPort);
    return CMkGbEndUser::Start(Param);
}

Uint32 CMkGbMediaEndUser::Play(const MkString& SipUri, BOOL bTcpSip, BOOL bTcpRtp, Uint32 WndIndex, const MkMediaStreamResultCallback& ResultCb, const MkStreamStoppedCallback& StreamStoppedCb, const MkEsPacketCallback& EsPacketCb)
{
    MkRtpParam OfferRtpParam;
    OfferRtpParam.bTcp = bTcpRtp;
    OfferRtpParam.Host = GetTcpPublicHost();
    OfferRtpParam.Ssrc = CMkUtil::GetRandomNumber();
    OfferRtpParam.Payload = 96;
    //only support rtp over tcp active,tcp client port can use any port
    CMkRtpUdpServer *pFreeUdpServer = GetFreeRtpUdpServer();
    if (!pFreeUdpServer) {
        return SystemBusy;
    }
    OfferRtpParam.Port = pFreeUdpServer->GetRtpPort();
    return CMkGbEndUser::StartPlay(SipUri, bTcpSip, WndIndex, OfferRtpParam, [this, ResultCb, EsPacketCb, pFreeUdpServer](Uint32 ErrorCode, const MkString& SessionId, const MkRtpParam& AnswerRtpParam, const EventResponseCallback& EventCb) {
        OnPlayResult(ResultCb, pFreeUdpServer, ErrorCode, SessionId, AnswerRtpParam, EsPacketCb, EventCb);
    }, StreamStoppedCb);
}

Uint32 CMkGbMediaEndUser::Play(const MkString& ChannelId, const MkString& PlayType, const CMkTime& BeginTime, const CMkTime& EndTime, Uint32 WndIndex, const MkMediaStreamResultCallback& ResultCb, const MkStreamStoppedCallback& StreamStoppedCb, const MkEsPacketCallback& EsPacketCb)
{
    return NoneError;
}

Uint32 CMkGbMediaEndUser::Publish(const MkString& SipUri, BOOL bTcpSip, BOOL bTcpRtp, CMkMediaSource* pMediaSource, const MkMediaStreamResultCallback& ResultCb, const MkStreamStoppedCallback& StreamStoppedCb)
{
    MkRtpParam OfferRtpParam;
    OfferRtpParam.bTcp = bTcpRtp;
    OfferRtpParam.Host = GetTcpPublicHost();
    OfferRtpParam.Ssrc = CMkUtil::GetRandomNumber();
    OfferRtpParam.Payload = 96;
    //only support rtp over tcp active,tcp client port can use any port
    CMkRtpUdpServer *pFreeUdpServer = GetFreeRtpUdpServer();
    if (!pFreeUdpServer) {
        return SystemBusy;
    }
    OfferRtpParam.Port = pFreeUdpServer->GetRtpPort();
    return CMkGbEndUser::StartPublish(SipUri, bTcpSip, OfferRtpParam, [this, ResultCb, pFreeUdpServer, pMediaSource](Uint32 ErrorCode, const MkString& SessionId, const MkRtpParam& AnswerRtpParam, const EventResponseCallback& EventCb) {
        OnPublisherResult(pMediaSource, ResultCb, pFreeUdpServer, ErrorCode, SessionId, AnswerRtpParam, EventCb);
    }, StreamStoppedCb);
}


Uint32 CMkGbMediaEndUser::Publish2(const MkString& SipUri, BOOL bTcpSip, BOOL bTcpRtp, CMkMediaSource* pMediaSource, const MkMediaStreamResultCallback& ResultCb, const MkStreamStoppedCallback& StreamStoppedCb)
{
    MkRtpParam OfferRtpParam;
    OfferRtpParam.bTcp = bTcpRtp;
    OfferRtpParam.Host = GetTcpPublicHost();
    OfferRtpParam.Ssrc = CMkUtil::GetRandomNumber();
    OfferRtpParam.Payload = 96;
    //only support rtp over tcp active,tcp client port can use any port
    CMkRtpUdpServer *pFreeUdpServer = GetFreeRtpUdpServer();
    if (!pFreeUdpServer) {
        return SystemBusy;
    }
    OfferRtpParam.Port = pFreeUdpServer->GetRtpPort();
    return CMkGbEndUser::StartPublish2(SipUri, bTcpSip, OfferRtpParam, [this, ResultCb, pFreeUdpServer, pMediaSource](Uint32 ErrorCode, const MkString& SessionId, const MkRtpParam& AnswerRtpParam, const EventResponseCallback& EventCb) {
        OnPublisherResult(pMediaSource, ResultCb, pFreeUdpServer, ErrorCode, SessionId, AnswerRtpParam, EventCb);
    }, StreamStoppedCb);
}

CMkRtpUdpServer* CMkGbMediaEndUser::GetFreeRtpUdpServer()
{
    Uint32 nMaxScore = 0;
    CMkRtpUdpServer* pUdpServer = nullptr;
    MkVector<CMkRtpUdpServer*> ::iterator it = m_VecRtpUdpServer.begin();
    for (; it != m_VecRtpUdpServer.end(); it++) {
        if ((*it)->GetScore() > nMaxScore) {
            pUdpServer = *it;
            nMaxScore = pUdpServer->GetScore();
        }
    }
    return pUdpServer;
}

Uint32 CMkGbMediaEndUser::OnPublisherResult(CMkMediaSource* pMediaSource, const MkMediaStreamResultCallback& ResultCb, CMkRtpUdpServer *pRtpUdpServer, Uint32 ResultErrorCode, const MkString& SessionId, const MkRtpParam& AnswerRtpParam, const EventResponseCallback& EventCb)
{
    if (NoneError != ResultErrorCode) {
        ResultCb(ResultErrorCode, SessionId);
        if (EventCb) {
            EventCb(MkMapString(), "");
        }
        return NoneError;
    }
    CMkRtp *pRtp = nullptr;
    if (!AnswerRtpParam.bTcp) {
        pRtp = pRtpUdpServer;
    } else {
        CMkRtpTcpClient *pRtpTcpClient = new CMkRtpTcpClient(GetEventPool()->GetFreeEvent(), GetTimer(), [this](CMkTcpClient *pClient) {
            pClient->DeleteLater();
        });
        pRtpTcpClient->Create(FALSE);
        pRtpTcpClient->ConnectServer(AnswerRtpParam.Host, AnswerRtpParam.Port, [this, EventCb]() {
            if (EventCb) {
                EventCb(MkMapString(), "");
            }
        });
        pRtp = pRtpTcpClient;
    }
    CMkGbConsumer *pConsumer = new CMkGbConsumer(AnswerRtpParam.Ssrc, pRtp);
    pConsumer->SetConsumerId(SessionId);
    MkRtpRemoter Remoter;
    Remoter.RtpPacketCb = nullptr;
    Remoter.RtpRemoteHost = AnswerRtpParam.Host;
    Remoter.RtpRemotePort = AnswerRtpParam.Port;
    Remoter.Ssrc = AnswerRtpParam.Ssrc;
    Remoter.Payload = AnswerRtpParam.Payload;
    Remoter.RtpType = MkRtpSender;
    Remoter.RtcpPacketCb = [this, SessionId, pMediaSource](const CMkRtcpPacket& RtcpPacket, const MkString& Host, Uint16 Port) {
        if (MkRtcpBye == RtcpPacket.GetPayload()) {
            BOOL bEmpty = FALSE;
            pMediaSource->DeleteConsumer(MkMediaConsumeGb, SessionId, bEmpty);
            ByeRequest(SessionId, nullptr);
            RemoveConsumer(SessionId);
        }
        return NoneError;
    };
    Remoter.RtpClosedCb = [this, SessionId, pMediaSource]() {
        BOOL bEmpty = FALSE;
        pMediaSource->DeleteConsumer(MkMediaConsumeGb, SessionId, bEmpty);
        ByeRequest(SessionId, nullptr);
        RemoveConsumer(SessionId);
    };
    pRtp->AddRtpRemoter(AnswerRtpParam.Ssrc, Remoter);
    pMediaSource->AddConsumer(new CMkGbConsume, pConsumer);
    m_GbConsumerLock.WLock();
    m_MapConsumer[SessionId] = pConsumer;
    m_GbConsumerLock.WUnlock();
    ResultCb(NoneError, SessionId);
    if (!AnswerRtpParam.bTcp&& EventCb) {
        EventCb(MkMapString(), "");
    }
    return NoneError;
}

Uint32 CMkGbMediaEndUser::OnPlayResult(const MkMediaStreamResultCallback& ResultCb, CMkRtpUdpServer *pRtpUdpServer, Uint32 ResultErrorCode, const MkString& SessionId, const MkRtpParam& AnswerRtpParam, const MkEsPacketCallback& EsPacketCb, const EventResponseCallback& EventCb)
{
    if (NoneError != ResultErrorCode) {
        ResultCb(ResultErrorCode, SessionId);
        if (EventCb) {
            EventCb(MkMapString(), "");
        }
        return NoneError;
    }
    CMkRtp *pRtp = nullptr;
    if (!AnswerRtpParam.bTcp) {
        pRtp = pRtpUdpServer;
    } else {
        CMkRtpTcpClient *pRtpTcpClient = new CMkRtpTcpClient(GetEventPool()->GetFreeEvent(), GetTimer(), [this](CMkTcpClient *pClient) {
            pClient->DeleteLater();
        });
        pRtpTcpClient->Create(FALSE);
        pRtpTcpClient->ConnectServer(AnswerRtpParam.Host, AnswerRtpParam.Port, [this, EventCb]() {
            if (EventCb) {
                EventCb(MkMapString(), "");
            }
        });
        pRtp = pRtpTcpClient;
    }
    CMkPsStreamSource *pSource = new CMkPsStreamSource();
    pSource->StartMediaSource(nullptr, EsPacketCb);
    MkRtpRemoter Remoter;
    Remoter.RtpPacketCb = [this, pSource](const CMkRtpPacket& RtpPacket, const MkString& Host, Uint16 Port) {
        pSource->InputDataSlice(RtpPacket.GetPayloadData(), RtpPacket.GetPayloadLength(), RtpPacket.GetMaker());
        return NoneError;
    };
    Remoter.RtpRemoteHost = AnswerRtpParam.Host;
    Remoter.RtpRemotePort = AnswerRtpParam.Port;
    Remoter.Ssrc = AnswerRtpParam.Ssrc;
    Remoter.Payload = AnswerRtpParam.Payload;
    Remoter.RtpType = MkRtpRecver;
    Remoter.RtcpPacketCb = [this, SessionId](const CMkRtcpPacket& RtcpPacket, const MkString& Host, Uint16 Port) {
        if (MkRtcpBye == RtcpPacket.GetPayload()) {
            ByeRequest(SessionId, nullptr);
            RemovePlayer(SessionId);
        }
        return NoneError;
    };
    Remoter.RtpClosedCb = [this, SessionId]() {
        ByeRequest(SessionId, nullptr);
        RemovePlayer(SessionId);
    };
    pRtp->AddRtpRemoter(AnswerRtpParam.Ssrc, Remoter);
    m_GbPlayLock.WLock();
    m_MapGbPlayer[SessionId] = pSource;
    m_GbPlayLock.WUnlock();
    ResultCb(NoneError, SessionId);
    if (!AnswerRtpParam.bTcp && EventCb) {
        EventCb(MkMapString(), "");
    }
    return NoneError;
}

Uint32 CMkGbMediaEndUser::RemoveConsumer(const MkString& SessionId)
{
    m_GbConsumerLock.WLock();
    MkMap<MkString, CMkGbConsumer*>::iterator it = m_MapConsumer.find(SessionId);
    if (it != m_MapConsumer.end()) {
        MkDelete(it->second);
        m_MapConsumer.erase(it);
    }
    m_GbConsumerLock.WUnlock();
    return NoneError;
}

Uint32 CMkGbMediaEndUser::RemovePlayer(const MkString& SessionId)
{
    m_GbPlayLock.WLock();
    MkMap<MkString, CMkPsStreamSource*>::iterator it = m_MapGbPlayer.find(SessionId);
    if (it != m_MapGbPlayer.end()) {
        MkDelete(it->second);
        m_MapGbPlayer.erase(it);
    }
    m_GbPlayLock.WUnlock();
    return NoneError;
}

Uint32 CMkGbMediaEndUser::Publish(const MkString& ChannelId, CMkMediaSource* pMediaSource, const MkMediaStreamResultCallback& ResultCb, const MkStreamStoppedCallback& StreamStoppedCb)
{
    return NoneError;
}

CMkGbMediaEndUser::~CMkGbMediaEndUser()
{
    MkMap<MkString, CMkGbConsumer*>::iterator it = m_MapConsumer.begin();
    for (; it != m_MapConsumer.end(); it++) {
        MkDelete(it->second);
    }
    m_MapConsumer.clear();

    MkMap<MkString, CMkPsStreamSource*>::iterator itPlayer = m_MapGbPlayer.begin();
    for (; itPlayer != m_MapGbPlayer.end(); itPlayer++) {
        MkDelete(itPlayer->second);
    }
    m_MapGbPlayer.clear();

    MkVector<CMkRtpUdpServer*>::iterator itRtp = m_VecRtpUdpServer.begin();
    for (; itRtp != m_VecRtpUdpServer.end(); itRtp++) {
        MkDelete(*itRtp);
    }
    m_VecRtpUdpServer.clear();
}